package flinkdemo.kafak;


import com.arcsoft.face.toolkit.ImageFactory;
import com.arcsoft.face.toolkit.ImageInfo;
import flinkdemo.kafak.entity.VideoMsg;
import net.coobird.thumbnailator.Thumbnails;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.io.ByteArrayInputStream;

/**
 * 保存了两份 图片数据 一个是原始图片 一个是图片的rgb数据，这个不合理。后续优化。
 * 保存原始图片是为了查询到后在原始图片上的人脸画框
 * @author ShiQiang
 */
public class ImgProcessingFlatMap extends RichFlatMapFunction<VideoMsg, Tuple2<String, VideoMsg>> {

    @Override
    public void flatMap(VideoMsg msg, Collector<Tuple2<String, VideoMsg>> collector) throws Exception {
        ByteArrayInputStream stream = new ByteArrayInputStream(msg.getImg());
        ImageInfo info = ImageFactory.bufferedImage2ImageInfo(Thumbnails.of(stream).scale(1).outputFormat("jpg").asBufferedImage());
        msg.setRgb(info.getImageData());
        msg.setWidth(info.getWidth());
        msg.setHeight(info.getHeight());
        collector.collect(new Tuple2<>(msg.getCameraId(),msg));
    }
}